package cn.itcast.tags.models.ml

import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.{MLModelTools, TagTools}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{VectorAssembler, VectorIndexer, VectorIndexerModel}
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit, TrainValidationSplitModel}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{Column, DataFrame, Dataset, Row, SparkSession}

/**
 * 用户购物性别
 */
class UsgTagModel extends AbstractModel("用户购物性别USF",ModelType.ML) {
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    val session: SparkSession = businessDF.sparkSession
    import session.implicits._
    /*root
          |-- cordersn: string (nullable = true)
          |-- ogcolor: string (nullable = true)
          |-- producttype: string (nullable = true)*/

    // 1. 获取订单表数据tbl_orders，与订单商品表数据关联获取会员ID
    val ordersDF: DataFrame = spark.read
      .format("hbase")
      .option("zkHosts", "bigdata-cdh01.itcast.cn")
      .option("zkPort", "2181")
      .option("hbaseTable", "tbl_tag_orders")
      .option("family", "detail")
      .option("selectFields", "memberid,ordersn")
      .load()

    ordersDF.show(10, false)

    // 2. 加载维度表数据：tbl_dim_colors（颜色）、tbl_dim_products（产品）
    // 2.1 加载颜色维度表数据
    val colorsDF: DataFrame = {
      spark.read
        .format("jdbc")
        .option("driver", "com.mysql.jdbc.Driver")
        .option("url", "jdbc:mysql://bigdata-cdh01.itcast.cn:3306/?useUnicode=true&characterEncoding=UTF8&serverTimezone=UTC")
        .option("dbtable", "profile_tags.tbl_dim_colors")
        .option("user", "root")
        .option("password", "123456")
        .load()
    }
    colorsDF.show(10, false)
    // 2.2. 构建颜色WHEN语句
    val colorColumn: Column = {
      // 声明变量
      var colorCol: Column = null
      colorsDF
        .as[(Int, String)].rdd
        .collectAsMap()
        .foreach { case (colorId, colorName) =>
          if (null == colorCol) {
            colorCol = when($"ogcolor".equalTo(colorName), colorId)
          } else {
            colorCol = colorCol.when($"ogcolor".equalTo(colorName), colorId)
          }
        }
      colorCol = colorCol.otherwise(0).as("color")
      // 返回
      colorCol
    }


    // 2.3 加载商品维度表数据
    val productsDF: DataFrame = {
      spark.read
        .format("jdbc")
        .option("driver", "com.mysql.jdbc.Driver")
        .option("url",
          "jdbc:mysql://bigdata-cdh01.itcast.cn:3306/?useUnicode=true&characterEncoding=UTF8&serverTimezone=UTC")
        .option("dbtable", "profile_tags.tbl_dim_products")
        .option("user", "root")
        .option("password", "123456")
        .load()
    }
    productsDF.show(10, false)

    // 2.4. 构建颜色WHEN语句
    var productColumn: Column = {
      // 声明变量
      var productCol: Column = null
      productsDF
        .as[(Int, String)].rdd
        .collectAsMap()
        .foreach { case (productId, prodcutName) =>
          if (null == productCol) {
            productCol = when($"producttype".equalTo(prodcutName), productId)
          } else {
            productCol =
              productCol.when(col("producttype").equalTo(prodcutName), productId)
          }
        }
      productCol = productCol.otherwise(0).as("product")
      productCol
    }

    // 根据运营规则标注的部分数据
    val labelColumn: Column = {
      when($"ogcolor".equalTo("樱花粉")
        .or($"ogcolor".equalTo("粉色"))
        .or($"ogcolor".equalTo("白色"))
        .or($"ogcolor".equalTo("香槟色"))
        .or($"ogcolor".equalTo("香槟金"))
        .or($"productType".equalTo("料理机"))
        .or($"productType".equalTo("挂烫机"))
        .or($"productType".equalTo("吸尘器/除螨仪")), 1) //女
        .otherwise(0) //男
        .as("label") //决策树预测label
    }

    //关联orders数据，颜色维度和商品类别维度
    val goodsDF: DataFrame = businessDF
      .join(ordersDF, businessDF("cordersn") === ordersDF("ordersn"))
      .select(
        $"memberid".as("userId"),
        colorColumn, //颜色
        productColumn, //产品类别
        labelColumn
      )

    //goodsDF.printSchema()
    /*
           |-- userId: string (nullable = true)
           |-- color: integer (nullable = false)
           |-- product: integer (nullable = false)
           |-- label: integer (nullable = false)
     */

    //实际开发使用标注数据,构建算法模型
    //val featuresDF: DataFrame = featuresTransform(goodsDF)
    //val dtcModel: DecisionTreeClassificationModel = trainModel(featuresDF)
    //val predictionDF: DataFrame = dtcModel.transform(featuresDF)

    //TODO 使用数据集训练管道模型
    //val pipelineModel: PipelineModel = trainPipelineModel(goodsDF)
    //val predictionDF: DataFrame = pipelineModel.transform(goodsDF)

    //TODO 使用训练验证分割方式，设置超参数训练处最佳模型
    //val pipelineModel: PipelineModel = trainBestModel(goodsDF)
    //val predictionDF: DataFrame = pipelineModel.transform(goodsDF)


    //TODO 使用交叉验证方式，设置超参数训练获取最佳模型
    val pipelineModel: PipelineModel = MLModelTools.loadModel(
      goodsDF,"usg",this.getClass).asInstanceOf[PipelineModel]
    val predictionDF: DataFrame = pipelineModel.transform(goodsDF)

    predictionDF.show(100, false)
    //按照用户ID进行分组，统计每个用户购物男性或女性个数及占比
    val genderDF: DataFrame = predictionDF
      .groupBy($"userId")
      .agg(
        count($"userId").as("total"),
        //判断label为0时，表示男性商品，设置为1，使用sum累加
        sum(
          when($"prediction".equalTo(0), 1).otherwise(0)
        ).as("maleTotal"),
        sum(
          when($"prediction".equalTo(1), 1).otherwise(0)
        ).as("femaleTotal")
      )
    //|userId|total|maleTotal|femaleTotal|

    //计算标签，
    //获取属性标签
    val rulesMap: Map[String, String] = TagTools.convertMap(tagDF)
    val rulesBroadcast: Broadcast[Map[String, String]] = session.sparkContext.broadcast(rulesMap)
    // 对每个用户，分别计算男性商品和女性商品占比，当占比大于0.6时，确定购物性别
    val gender_tag_udf: UserDefinedFunction = udf(
      (total: Long, maleTotal: Long, femaleTotal: Long) => {
        val maleRate: Double = maleTotal / total.toDouble
        val femaleRate: Double = femaleTotal / total.toDouble
        if (maleRate >= 0.6) {
          rulesBroadcast.value("0")
        } else if (femaleRate >= 0.6) {
          rulesBroadcast.value("1")
        } else {
          rulesBroadcast.value("-1")
        }
      }
    )

    //获取画像标签数据
    val modelDF: DataFrame = genderDF.select(
      $"userId",
      gender_tag_udf($"total", $"maleTotal", $"femaleTotal").as("usg")
    )

    modelDF.show(100, false)


    modelDF
  }

  /**
   * 针对数据集进行特征工程：特征提取、特征转换及特征选择
   *
   * @param dataframe 数据集
   * @return 数据集，包含特征列features: Vector类型和标签列label
   */
  def featuresTransform(dataframe: DataFrame): DataFrame = {
    // a. 特征向量化
    val assembler: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("color", "product"))
      .setOutputCol("raw_features")
    val df1: DataFrame = assembler.transform(dataframe)
    // b. 类别特征进行索引
    val vectorIndexer: VectorIndexerModel = new VectorIndexer()
      .setInputCol("raw_features")
      .setOutputCol("features")
      .setMaxCategories(30) //最大类别
      .fit(df1)
    val df2: DataFrame = vectorIndexer.transform(df1)
    // c. 返回特征数据
    df2
  }

  /**
   * 使用决策树分类算法训练模型，返回DecisionTreeClassificationModel模型
   *
   * @return
   */
  def trainModel(dataframe: DataFrame): DecisionTreeClassificationModel = {
    // a. 数据划分为训练数据集和测试数据集
    val Array(trainingDF, testingDF) = dataframe.randomSplit(Array(0.8, 0.2), seed = 123L)
    // b. 构建决策树分类器
    val dtc: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMaxDepth(5) // 树的深度
      .setMaxBins(32) // 树的叶子数目
      .setImpurity("gini") // 基尼系数
    // c. 训练模型
    logWarning("正在训练模型...................................")
    val dtcModel: DecisionTreeClassificationModel = dtc.fit(trainingDF)
    // d. 模型评估
    val predictionDF: DataFrame = dtcModel.transform(testingDF)
    println(s"accuracy = ${modelEvaluate(predictionDF, "accuracy")}")
    // e. 返回模型
    dtcModel
  }

  /**
   * 模型评估，返回计算分类指标值
   *
   * @param dataframe  预测结果的数据集
   * @param metricName 分类评估指标名称，支持：f1、weightedPrecision、weightedRecall、accuracy
   * @return
   */
  def modelEvaluate(dataframe: DataFrame, metricName: String): Double = {
    // a. 构建多分类分类器
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      // 指标名称，
      .setMetricName(metricName)
    // b. 计算评估指标
    val metric: Double = evaluator.evaluate(dataframe)
    // c. 返回指标
    metric
  }

  /**
   * 使用决策树分类算法训练模型，返回PipelineModel模
   */
  def trainPipelineModel(dataframe: DataFrame): PipelineModel = {
    //userId color product label
    val Array(trainingDF, testingDF) = dataframe.randomSplit(Array(0.8, 0.2), 123)
    // a. 特征向量化
    val assembler: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("color", "product"))
      .setOutputCol("raw_features")
    // b. 类别特征进行索引
    val vectorIndexer: VectorIndexer = new VectorIndexer()
      .setInputCol("raw_features")
      .setOutputCol("features")
      .setMaxCategories(30) //最大类别

    // c. 构建决策树分类器
    val dtc: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMaxDepth(5) // 树的深度
      .setMaxBins(32) // 树的叶子数目
      .setImpurity("gini") // 基尼系数

    val pipeline: Pipeline = new Pipeline()
      .setStages(Array(assembler, vectorIndexer, dtc))

    val pipelineModel: PipelineModel = pipeline.fit(trainingDF)

    //模型评估
    val predictionDF: DataFrame = pipelineModel.transform(testingDF)
    predictionDF.show(100, false)
    println(s"accuracy=${modelEvaluate(predictionDF,"accuracy")}")
    pipelineModel
  }

  /**
   * 调整算法超参数，找出最优模型
   * @param dataframe 数据集
   * @return
   */
  def trainBestModel(dataframe: DataFrame): PipelineModel = {
    //userId color product label
    // a. 特征向量化
    val assembler: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("color", "product"))
      .setOutputCol("raw_features")
    // b. 类别特征进行索引
    val vectorIndexer: VectorIndexer = new VectorIndexer()
      .setInputCol("raw_features")
      .setOutputCol("features")
      .setMaxCategories(30) //最大类别

    // c. 构建决策树分类器
    val dtc: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMaxDepth(5) // 树的深度
      .setMaxBins(32) // 树的叶子数目
      .setImpurity("gini") // 基尼系数

    val pipeline: Pipeline = new Pipeline()
      .setStages(Array(assembler, vectorIndexer, dtc))

    //创建一个网格参数对象实例
    val paramGrid: Array[ParamMap] = new ParamGridBuilder()
      .addGrid(dtc.maxDepth, Array(5, 10))
      .addGrid(dtc.impurity, Array("gini", "entropy"))
      .addGrid(dtc.maxBins, Array(32, 64))
      .build()

    //创建训练验证分割实例对象，设置算法
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      // 指标名称，支持：f1、weightedPrecision、weightedRecall、accuracy
      .setMetricName("accuracy")


    // g. 训练验证
    val trainValidationSplit = new TrainValidationSplit()
      .setEstimator(pipeline) //算法，
      .setEvaluator(evaluator) //评估其
      .setEstimatorParamMaps(paramGrid) //算法超参数
      // 80% of the data will be used for training and the remaining 20% for validation.
      .setTrainRatio(0.8)

    // h. 训练模型
    val splitModel: TrainValidationSplitModel = trainValidationSplit.fit(dataframe)

    //获取最佳模型
    val pipelineModel: PipelineModel = splitModel.bestModel.asInstanceOf[PipelineModel]
    pipelineModel
  }
}
object UsgTagModel{
  def main(args: Array[String]): Unit = {
    val model = new UsgTagModel
    model.executeModel(378L,false)
  }
}

